package com.example.factory;

import com.example.model.SysJobInfo;
import com.example.service.SysJobInfoService;
import com.example.service.SysJobLogService;
import com.example.transport.TransportService;
import com.exmaple.dto.RpcRequest;
import com.exmaple.dto.RpcResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Service
public class JobSchedulerFactory implements InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass());

    private LinkedBlockingQueue<RpcResponse> rpcResponseQueue = new LinkedBlockingQueue<>();

    private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

    private volatile static boolean started = false;

    private Thread timeWheelThread = null;

    private Thread scheduleThread = null;

    private Thread rpcResponseThread = null;

    public static final long PRE_READ_MS = 5000;

    @Resource
    private SysJobInfoService sysJobInfoService;

    @Resource
    private SysJobLogService sysJobLogService;

    @Resource
    private TransportService transportService;

    @Resource
    private DataSource dataSource;

    @Override
    public void destroy() throws Exception {
        started = false;
        if (this.timeWheelThread.isAlive()) {
            this.timeWheelThread.interrupt();
        }
        if (this.scheduleThread.isAlive()) {
            this.scheduleThread.interrupt();
        }
        if (this.rpcResponseThread.isAlive()) {
            this.rpcResponseThread.interrupt();
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.timeWheelThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!started) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
                        int nowSecond = Calendar.getInstance().get(Calendar.SECOND);
                        List<Integer> ringItemData = new ArrayList<>();
                        for (int i = 0; i < 2; i++) {
                            List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
                            if (tmpData != null) {
                                ringItemData.addAll(tmpData);
                            }
                        }
                        if (ringItemData.size() > 0) {
                            for (int jobId : ringItemData) {
                                RpcRequest rpcRequest = sysJobInfoService.buildRequest(jobId);
                                RpcResponse rpcResponse = transportService.handler(rpcRequest);
                                sysJobLogService.addLog(rpcResponse);
                            }
                            ringItemData.clear();
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        this.timeWheelThread.setName("timeWheelThread");
        this.timeWheelThread.setDaemon(true);
        this.timeWheelThread.start();
        this.scheduleThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    TimeUnit.MILLISECONDS.sleep(PRE_READ_MS - System.currentTimeMillis() % 1000);
                    logger.warn(">>>>>>>>> scheduleThread started ...");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                while (!started) {
                    Connection conn = null;
                    Boolean connAutoCommit = null;
                    PreparedStatement preparedStatement = null;
                    try {
                        conn = dataSource.getConnection();
                        connAutoCommit = conn.getAutoCommit();
                        conn.setAutoCommit(false);

                        preparedStatement = conn.prepareStatement("select * from SYS_JOB_LOCK where id = 1 for update");
                        preparedStatement.execute();


                        long nowTime = System.currentTimeMillis();
                        List<SysJobInfo> sysJobInfos = sysJobInfoService.findListByTime(nowTime + PRE_READ_MS);
                        if (sysJobInfos != null && sysJobInfos.size() > 0) {
                            for (SysJobInfo sysJobInfo : sysJobInfos) {
                                try {
                                    if (nowTime > sysJobInfo.getJobRunTime() + PRE_READ_MS) {
                                        sysJobInfoService.refreshRunTime(sysJobInfo.getId(), System.currentTimeMillis());
                                    } else if (nowTime > sysJobInfo.getJobRunTime()) {
                                        sysJobInfoService.refreshRunTime(sysJobInfo.getId(), System.currentTimeMillis());
                                        if (sysJobInfo.getEnabled() && nowTime + PRE_READ_MS > sysJobInfo.getJobRunTime()) {
                                            int ringSecond = (int) ((sysJobInfo.getJobRunTime() / 1000) % 60);
                                            pushTimeRing(ringSecond, sysJobInfo.getId());
                                            sysJobInfoService.refreshRunTime(sysJobInfo.getId(), sysJobInfo.getJobRunTime());
                                        }
                                    } else {
                                        int ringSecond = (int) ((sysJobInfo.getJobRunTime() / 1000) % 60);
                                        pushTimeRing(ringSecond, sysJobInfo.getId());
                                        sysJobInfoService.refreshRunTime(sysJobInfo.getId(), sysJobInfo.getJobRunTime());
                                    }
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            }
                        }

                    } catch (Exception e) {
                        logger.warn("scheduleThread error " + e.getMessage());
                    } finally {
                        if (conn != null) {
                            try {
                                conn.commit();
                            } catch (SQLException e) {
                                e.printStackTrace();
                            }
                            try {
                                conn.setAutoCommit(connAutoCommit);
                            } catch (SQLException e) {
                                e.printStackTrace();
                            }
                            try {
                                conn.close();
                            } catch (SQLException e) {
                                e.printStackTrace();
                            }
                        }
                        if (null != preparedStatement) {
                            try {
                                preparedStatement.close();
                            } catch (SQLException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }
            }
        });
        this.scheduleThread.setName("scheduleThread");
        this.scheduleThread.setDaemon(true);
        this.scheduleThread.start();
        this.rpcResponseThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!started) {
                    try {
                        RpcResponse rpcResponse = rpcResponseQueue.take();
                        sysJobLogService.saveRpcResponse(rpcResponse);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        this.rpcResponseThread.setName("rpcResponseThread");
        this.rpcResponseThread.setDaemon(true);
        this.rpcResponseThread.start();
    }

    private void pushTimeRing(int ringSecond, int jobId) {
        List<Integer> ringItemData = ringData.get(ringSecond);
        if (ringItemData == null) {
            ringItemData = new ArrayList<Integer>();
            ringData.put(ringSecond, ringItemData);
        }
        ringItemData.add(jobId);
    }
}
